package core.sql.udf;

import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SparkSession;
import org.apache.spark.sql.api.java.UDF1;
import org.apache.spark.sql.types.DataType;
import org.apache.spark.sql.types.DataTypes;
import org.apache.spark.sql.types.StringType$;

public class Spark02_UDF {

    public static void main(String[] args) {
        SparkSession sparkSession = SparkSession
                .builder()
                .master("local[*]")
                .appName("Spark01_UDF")
                .getOrCreate();

        try {
            // 读取 JSON 数据
            Dataset<Row> json = sparkSession.read()
                    .option("mode", "PERMISSIVE") // 允许坏记录
                    .option("badRecordsPath", "data/bad_records") // 指定错误记录的输出路径
                    .json("data/user_sql.json");


            /**
             * register需要传递3个参数
             * 参数1：SQL中使用的方法名
             * 参数2：逻辑
             * 参数3：数据类型,需要特殊的使用方式，需要用scalla语法进行操作StringType$.MODULE$。或者以工具类的形式DataTypes.StringType
             */
            sparkSession.udf().register("perfixName", new UDF1<String, String>() {
                @Override
                public String call(String input) throws Exception {
                    return "Name:"+input;
                }
            }, DataTypes.StringType);

            // 缓存解析后的结果
            json.cache();

            // 创建临时视图
            json.createOrReplaceTempView("user");

            // 示例：执行 SQL 查询
            Dataset<Row> result = sparkSession.sql("SELECT perfixName(name) as name FROM user");
            result.show(); // 显示结果

//            // 处理错误记录
//            Dataset<Row> corruptRecords = sparkSession.read().json("data/bad_records/*");
//            corruptRecords.show(); // 显示错误记录

        } catch (Exception e) {
            e.printStackTrace(); // 捕获并打印异常
        } finally {
            sparkSession.close();
        }
    }

}

